

package com.hazelcast.query.impl.getters;

import com.hazelcast.internal.namespace.NamespaceUtil;
import com.hazelcast.internal.serialization.Data;
import com.hazelcast.internal.serialization.InternalSerializationService;
import com.hazelcast.internal.serialization.impl.GenericRecordQueryReader;
import com.hazelcast.internal.serialization.impl.InternalGenericRecord;
import com.hazelcast.query.extractor.ValueExtractor;
import com.hazelcast.query.impl.DefaultValueCollector;

final class ExtractorGetter extends Getter {

    private final ValueExtractor extractor;
    private final Object arguments;
    private final InternalSerializationService serializationService;

    ExtractorGetter(InternalSerializationService serializationService, ValueExtractor extractor, Object arguments) {
        super(null);
        this.extractor = extractor;
        this.arguments = arguments;
        this.serializationService = serializationService;
    }

    @Override
    @SuppressWarnings("unchecked")
    Object getValue(Object target) throws Exception {
        Object extractionTarget;
        // This part will be improved in 3.7 to avoid extra allocation
        DefaultValueCollector collector = new DefaultValueCollector();
        if (target instanceof Data data) {
            InternalGenericRecord record = serializationService.readAsInternalGenericRecord(data);
            extractionTarget = new GenericRecordQueryReader(record);
        } else {
            extractionTarget = target;
        }
        NamespaceUtil.runWithOwnClassLoader(extractor, () -> extractor.extract(extractionTarget, arguments, collector));
        return collector.getResult();
    }

    @Override
    Class<?> getReturnType() {
        throw new UnsupportedOperationException();
    }

    @Override
    boolean isCacheable() {
        return true;
    }

}
